package com.burges.net.dataStream.codeRuler.transform.SingleDataStream

import org.apache.flink.api.common.functions.FlatMapFunction
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment, _}
import org.apache.flink.util.Collector

/**
  * 创建人    BurgessLee 
  * 创建时间   2020/1/26 
  * 描述     FlatMap操作
  */
object FlatMapFunctionDemo {

	def main(args: Array[String]): Unit = {
		val env = StreamExecutionEnvironment.getExecutionEnvironment

		val dataStream = env.fromElements("a,b,c", "d,f,e", "c,d,a", "c,b,d")
//		val flatMapStream = dataStream.flatMap( str => str.split(","))

		//使用方式一
		val resStream = dataStream.flatMap(_.split(","))

		//使用方式二
		//指定FlatMapFunction
		val flatMapStream_02: DataStream[String] = dataStream.flatMap(new FlatMapFunction[String, String] {
			override def flatMap(t: String, collector: Collector[String]): Unit = {
				t.split(",").foreach(collector.collect)
			}
		})

		resStream.print()
		flatMapStream_02.print()
		env.execute("Streaming FlatMapOperator")

	}

}
